package opc.read.sycread;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.SneakyThrows;
import opc.read.config.PathConf;
import opc.read.kafka.KafkaSelfProducer;
import opc.read.utils.*;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.jinterop.dcom.common.JIException;
import org.openscada.opc.lib.common.ConnectionInformation;
import org.openscada.opc.lib.common.NotConnectedException;
import org.openscada.opc.lib.da.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @Description: 输入信号-》opc转换一下-》输出信号，模拟仙鹤的opcserve信号，同步读取
 * @author: liyuan
 * @data 2021-01-06 10:21
 */
public class ByReadUseJson2 {
    private static final Logger logger = LoggerFactory.getLogger(ByReadUseJson2.class);

    public static void DemoAPI() throws IOException {
//        final SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");
        final SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd-HH");
// 读取json信号文件，得到字符串。将字符串转成LinkedHashMap
        LinkedHashMap<String, Object> xinhaoSources = ReadJSONItem.xianHe();

        Map<String, String> dataSources = ReadHostProt.trueHostProp();

        final ConnectionInformation ci = new ConnectionInformation();
        ci.setHost(dataSources.get("host"));
        ci.setDomain(dataSources.get("domain"));
        ci.setUser(dataSources.get("user"));
        ci.setPassword(dataSources.get("password"));
        ci.setClsid(dataSources.get("clsid"));

        final Server server = new Server(ci, Executors.newSingleThreadScheduledExecutor());
        final AutoReconnectController autoReconnectController = new AutoReconnectController(server);
        // 存放模拟或者说实际的信号源  输入和输出json的信号源的key是一样的
        final Map<String, Map<String, Item>> itemSource = new LinkedHashMap<String, Map<String, Item>>();
        // 存放结果
        final MyCopyOnWriteMap<String, Object> result = new MyCopyOnWriteMap<String, Object>(xinhaoSources);

        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);

        try {
            // connect to server
            autoReconnectController.connect(); //连接opcserver
            final AccessBase access = new SyncAccess(server, 1000); // 获取信号的对象，这里使用同步方式
//            final AccessBase access = new Async20Access(server, 1000, false); // 异步方式  // Failed to change state (true).   changed()方法不被调用，导致result一直不更新
            for (Map.Entry<String, Object> groupMap : xinhaoSources.entrySet()) {  // json源信号
                String groupName = groupMap.getKey();
                if (groupName.equals("date")) {
                    continue;
                }
                Group group = null; // 这里有个坑，写  Group group =server.addGroup(groupName) ;会连接不到
                group = server.addGroup(groupName);
                group.setActive(true);
                List<String> shiPeiItems = new ArrayList<String>();//适配信号
                for (String itemName : ((Map<String, Object>) groupMap.getValue()).keySet()) {
                    if (itemName.startsWith("PM28_CN.") || itemName.startsWith("PM28_ZQ.") || itemName.startsWith("PM28_BJQS.") || itemName.startsWith("PM28_ZZQS.") || itemName.startsWith("PM28_GAS_Z.")
                            || itemName.startsWith("PM28_FT28878B.") || itemName.startsWith("PM28_JF.") || itemName.startsWith("PM28_GF.") || itemName.startsWith("PM28_DG.")) {
                        itemName = itemName.replace(".", "__");
                        shiPeiItems.add("xianhe.设备 1." + itemName);
                    } else {
                        shiPeiItems.add("xianhe.设备 1." + itemName.split("\\.")[0]);
                    }
                }
                // 得到信号数组
                String[] items = shiPeiItems.toArray(new String[shiPeiItems.size()]);
                Map<String, Item> itemMap = group.addItems(items);
                // itemSource 包含组、组信号，是输入到opcserver的信号
                itemSource.put(groupName, itemMap);
            }
            final Map<String, Long> times = new LinkedHashMap<String, Long>();
            final AtomicBoolean timeFlag = new AtomicBoolean(true);
            //首先遍历读一遍
            result.put("date", System.currentTimeMillis());
            /***输入到opcserver的信号值 **/
            for (final String groupName : itemSource.keySet()) {
                Map<String, Item> itemMap = itemSource.get(groupName);
                //组内部的读取信号
                for (Map.Entry<String, Item> map : itemMap.entrySet()) {
                    if (map.getValue() instanceof Item) {
                        Item item = map.getValue();
                        Object object = item.read(false).getValue().getObject();
                        Short quality = item.read(false).getQuality();
                        // 当信号质量值小于192的时候，表示该信号有问题，该信号值赋值为null
                        if (quality < 192) {
                            object = null;
                        }
                        // 去掉xianhe.设备 1., 只留下需要的信号名
                        String s = item.getId().split("\\.")[2];
                        /*** 从opcserver输出的信号值 转换一下 存到result中 **/
                        String itemKey = ItemKey2Convert.in2Out(s);
                        Map<String, Object> itemMap2 = (Map<String, Object>) result.get(groupName);
                        if (itemMap2.containsKey(itemKey)) {
                            itemMap2.put(itemKey, object);
                        } else {
                            logger.error("键值对出错：{},{}", s, itemKey);
                        }
                        //计算采样时间的测试代码时间
                        if (PathConf.isTestEnv) {
                            if (item.getId().equals("xianhe.设备 1.PT28851A")) {
                                times.put("PT28851A" + times.size(), System.currentTimeMillis());
                            } else if (item.getId().equals("xianhe.设备 1.PM28_SET_WIDTH")) {
                                times.put("PM28_SET_WIDTH" + times.size(), System.currentTimeMillis());
                            }
                        }

                    }
                }
            }

            final Producer producer = KafkaSelfProducer.getProducer();
            final Gson gson = new GsonBuilder().enableComplexMapKeySerialization().create();
            //新开线程开始读取
            scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @SneakyThrows
                @Override
                public void run() {
                    try {
                        result.put("date", System.currentTimeMillis());
                        /***格式化之后没有按照存入的顺序，所以使用gson***/
//                    String jsonString = JSON.toJSONString(result);
                        String jsonString = gson.toJson(result);
                        logger.info("--json---\n{}", jsonString);
                        logger.info("将结果保存到文本文件中...");
//                        WriterFile.writeTxtFile(jsonString, new File(PathConf.filePath + formatter.format(new Date()) + ".txt"), "utf-8");
                        logger.info("发送数据到kafka...");
                        producer.send(new ProducerRecord<String, String>("zhenxing0", jsonString));
                        /** future会出现异常的 Future<RecordMetadata> future =  **/
                        if (PathConf.isTestEnv) {
                            //                        RecordMetadata zhenxing0 = future.get();
//                        logger.info("send to topic：" + zhenxing0.topic() + "partition：" + zhenxing0.partition() + "offset：" + zhenxing0.offset());
                            if (times.size() >= 10 && timeFlag.get() == true) { /***用于计算采集一遍需要的时间 **/
                                WriterFile.writeTxtFile(gson.toJson(times), new File(PathConf.timeJar + formatter.format(new Date()) + ".txt"), "utf-8");
                                timeFlag.set(false);
                            }
                        }
                    } catch (Exception e) {
                        logger.error("---终于异常了————————", e);
                        e.printStackTrace();
                    }
                }
            }, 2000, 500, TimeUnit.MILLISECONDS);//单位微妙

            // 因为有另外一个线程活着，所以这个线程会一直工作，不会退出。当没有起另外一个线程时，不加while(true)，程序就会退出
            for (final String groupName : itemSource.keySet()) {
                Map<String, Item> itemMap = itemSource.get(groupName);
                //组内部的读取信号
                for (Map.Entry<String, Item> map : itemMap.entrySet()) {
                    if (map.getValue() instanceof Item) {
                        //读取单个数据的代码
                        access.addItem(map.getValue().getId(), new DataCallback() {
                            /*** DataCallback()是信号值变化的事件回调函数  从结果来看，没有改变的信号也在输出  **/
                            @Override
                            public void changed(Item item, ItemState state) {
                                // Item和ItemState的区别
//                                try {
//                                    System.out.println("name: " + item.getId().split("\\.")[2] + " ,value: " + state.getValue().getObject());
//                                } catch (JIException e) {
//                                    e.printStackTrace();
//                                }
                                String tmpKey = item.getId().split("\\.")[2];
                                String itemKey = ItemKey2Convert.in2Out(tmpKey); /** 从opcserver出来的值转换一下 **/
                                Map<String, Object> itemMap = (Map<String, Object>) result.get(groupName);
                                try {
                                    if (itemMap.containsKey(itemKey)) {
                                        itemMap.put(itemKey, state.getValue().getObject());  // 去掉xianhe.设备 1., 只留下需要的信号名
//                                        System.out.println("itemKey: " + itemKey + ", state.getValue().getObject(): " + state.getValue().getObject());
                                    } else {
                                        logger.info("键值对出错：{},{}", tmpKey, itemKey);
                                    }
                                    if(PathConf.isTestEnv){
                                        if (times.size() < 10) {
                                            if (item.getId().equals("xianhe.设备 1.PT28851A")) {
                                                times.put("PT28851A" + times.size(), System.currentTimeMillis());
                                            } else if (item.getId().equals("xianhe.设备 1.PM28_SET_WIDTH")) {
                                                times.put("PM28_SET_WIDTH" + times.size(), System.currentTimeMillis());
                                            }
                                        }
                                    }

                                } catch (JIException e) {
                                    e.printStackTrace();
                                }

                            }
                        });
                    }
                }
            }
            access.bind();
//            Thread.sleep(100);
//            access.unbind();
        }
        catch (Exception e) {
//            UnknownHostException, NotConnectedException, JIException, DuplicateGroupException
            if(e instanceof NotConnectedException){
                try{
                    logger.error("出现未连接异常，系统自启\n{}",e);
                    Thread.sleep(1000*60*5);//5分钟
                    String cmd="D:\\ProDataAccess\\ya\\xianheJar\\xianhe.bat";
                    Runtime.getRuntime().exec(cmd);
                    System.exit(0);
                }catch (Exception e2){
                    e2.printStackTrace();
                }
            }
            /** 其他地方异常则检查代码，没必要自拉起  **/
            logger.error("---大范围异常了----", e);
            e.printStackTrace();
        }
    }


}
